package net.spatula.pile;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.channels.FileChannel.MapMode;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Pile {

	private static final int INDEX_BUCKETS = 65536;
	private static final int OVERFLOW_BUCKETS = 1;
	private static final int BUCKET_FILES = 256;

	private static final int KEY_SIZE = 512 / 8;
	private static final int OFFSET_SIZE = 64 / 8;
	private static final int LENGTH_SIZE = 32 / 8;
	private static final int ENTRY_SIZE = KEY_SIZE + OFFSET_SIZE + LENGTH_SIZE;
	private static final int BUCKET_SIZE = ENTRY_SIZE * BUCKET_FILES;
	
	private static final String indexFileExt = ".pyx";
	private static final String pileFileExt = ".pyl";
	
	
	/*
	 * Pile Formats:
	 * 
	 * Index:
	 * Bytes 0-4: MAGIC
	 * Byte 5: VERSION
	 * Bytes 6-7: reserved
	 * int: count of allowed buckets
	 * int: number of allocated buckets
	 * int[65537]: ordering of index entries (initialized to all -1 for empty entries)
	 * Buckets
	 *  Entries
	 *    Bytes 0-63: key
	 *    long: index location into pile file
	 *    int: length of entry 
	 *
	 * Pile:
	 * File entries
	 *  int: offset to start of file data
	 *  byte[]: arbitrary meta-data
	 *  NULL
	 *  byte[]: file data
	 */
	
	private final MessageDigest digest; 
	
	private static final Logger LOGGER = LoggerFactory.getLogger(Pile.class);

	private RandomAccessFile pileFile;
	private RandomAccessFile pileIndex;
	
	private int[] orderIndex = new int[INDEX_BUCKETS + OVERFLOW_BUCKETS];

	public Pile(File location, boolean initialize) throws IOException {
		MessageDigest instance = null;
		try {
			instance = MessageDigest.getInstance("SHA-512");
		} catch (NoSuchAlgorithmException e) {
			
		}
		digest = instance;
		
		if (!location.exists() && !initialize) {
			throw new FileNotFoundException();
		}

		if (initialize) {
			pileFile = createNewPile(location);
		}
	}

	private RandomAccessFile createNewPile(File location) throws IOException {
		RandomAccessFile file = new RandomAccessFile(location, "rw");
		file.setLength(ENTRY_SIZE * (INDEX_BUCKETS + OVERFLOW_BUCKETS)
				* BUCKET_FILES);
		LOGGER.debug("Pile created, size is " + file.length());
		return file;
	}

	RandomAccessFile getPileFile() {
		return pileFile;
	}
	
	RandomAccessFile getPileIndex() {
		return pileIndex;
	}

	long getIndexLength() {
		return (INDEX_BUCKETS + OVERFLOW_BUCKETS) * BUCKET_FILES * ENTRY_SIZE;
	}

	public long updateItem(Item item) {
		long newLocation;
		try {
			newLocation = appendItem(item);
			int bucketNumber = calculateBucketNumber(item.getPath());
			updateBucket(bucketNumber, item, newLocation, true);
			return newLocation;
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 
		return -1;
	}

	private void updateBucket(int bucketNumber, Item item, long newLocation, boolean append) throws IOException, FileNotFoundException {
		int location = bucketNumber * BUCKET_SIZE;
		FileChannel channel;
		FileLock lock = null;
		byte[] hash = hash(item.getPath());
		try {
			channel = pileFile.getChannel();
			LOGGER.info("Locking bucket, channel=" + channel.hashCode());
			lock = channel.lock(location, BUCKET_SIZE, false);
			
			LOGGER.debug("Mapping bucket to memory");
			MappedByteBuffer map = channel.map(MapMode.READ_WRITE, location,
					BUCKET_SIZE);

			byte[] thisKey = new byte[KEY_SIZE];
			byte[] zeros = new byte[KEY_SIZE];
			boolean found = false;
			int lastEntry = 0;
			for (int i = 0; i < BUCKET_FILES; i++) {
				map.position(i * ENTRY_SIZE);
				map.get(thisKey, 0, KEY_SIZE);
				if (Arrays.equals(thisKey, hash)) {
					found = true;
					LOGGER.debug("Found existing entry for item, updating");
					writeEntry(item, newLocation, map);
				} else if (Arrays.equals(thisKey, zeros))
				{
					lastEntry = i;
					break;
				}
			}
			if (!found)
			{
				if (append) {
				LOGGER.debug("Existing entry not found; appending new entry at " + lastEntry * ENTRY_SIZE);
				map.position(lastEntry * ENTRY_SIZE);
				map.put(hash);
				
				writeEntry(item, newLocation, map);
				} else {
					throw new FileNotFoundException();
				}
			}
		} finally {
			if (lock != null) {
				lock.release();
			}
		}

	}

	private void writeEntry(Item item, long location, MappedByteBuffer map) {
		LOGGER.debug("Writing pointer at " + map.position());
		map.putLong(location);
		map.putInt(item.getData() == null ? 0 : item.getData().length);
		map.force();
	}

	protected int calculateBucketNumber(String path) {
		
		
		byte[] bytes = hash(path);

		// condense the 512 bits down to 16 bits, knowing that we will of course
		// have many more collisions
		int sum = 0;
		int bitNumber = 0;
		for (int i = 0; i < bytes.length; i += 4, bitNumber++) {
			byte thisByte = bytes[i];
			int bit = bitNumber % 8;
			int bitVal = (int) Math.pow(2, bit);
			sum <<= 1;
			if ((thisByte & bitVal) != 0)
			{
				sum |= 1;
			}
		}
		LOGGER.debug("Bucket number for " + path + " is " + (sum & 65535));
		return sum & 65535;
		
	}

	private byte[] hash(String path)  {
		
		try {
			digest.reset();
			byte[] bytes = digest.digest(path.getBytes("UTF-8"));
			LOGGER.debug(path + " hashed to " + Arrays.asList(bytes));
			return bytes;
		} catch (UnsupportedEncodingException e) {
			// do nothing; unrecoverable (should never happen anyway)
		}
	
		return null;
		
	}

	private long appendItem(Item item) throws IOException {
		long eof = -1;
		FileChannel channel;
		FileLock lock = null;
		try {
			synchronized (pileFile) {
				eof = pileFile.length();
				channel = pileFile.getChannel();
				LOGGER.debug("Locking new segment");
				lock = channel.lock(eof, item.getData().length, false);
			}
			
			LOGGER.debug("Writing item");
			channel.write(ByteBuffer.wrap(item.getData()), eof);
		} finally {
			if (lock != null) {
				lock.release();
			}
		}
		return eof;
	}

	public byte[] find(String path) throws FileNotFoundException, IOException {
		
		FileChannel channel;
		FileLock lock = null;
		

		try {
			byte[] hash = hash(path);
			int bucketNumber = calculateBucketNumber(path);
			int location = bucketNumber * BUCKET_SIZE;
			
			channel = pileFile.getChannel();
			LOGGER.info("Locking bucket; channel=" + channel.hashCode());
			lock = null; //channel.lock(location, BUCKET_SIZE, true);
			
			LOGGER.debug("Reading bucket to memory");
			ByteBuffer bucket = ByteBuffer.allocate(BUCKET_SIZE);
			channel.read(bucket, location);
			
			byte[] thisKey = new byte[64];
			
			for (int i = 0; i < BUCKET_FILES; i++) {
				bucket.position(i * ENTRY_SIZE);
				bucket.get(thisKey, 0, KEY_SIZE);
				if (Arrays.equals(thisKey, hash)) {
					
					LOGGER.debug("Found entry for item at index " + i);
					long offset = bucket.getLong();
					int length = bucket.getInt();
					LOGGER.debug("Contents reside at "+offset + "," + length);
					
					if (offset == 0 && length == 0)
					{
						// it has been flagged deleted
						break;
					}
					
					ByteBuffer buffer = ByteBuffer.allocate(length);
					channel.read(buffer, offset);
					buffer.rewind();
					
					byte[] data = new byte[length];
					buffer.get(data);
					return data;
				}
			}
		} finally {
			if (lock != null)
			{
				try {
					lock.release();
				} catch (IOException e) {
					// do nothing
				}
			}
		}
		
		throw new FileNotFoundException();
	}

	public void delete(String path) throws FileNotFoundException, IOException {
			int bucketNumber = calculateBucketNumber(path);
			updateBucket(bucketNumber, new Item(path, null), 0, false);
	}

}
